package com.wuwangfu.func;


import com.wuwangfu.entity.Behavior;
import org.apache.flink.api.common.functions.AggregateFunction;

/**
 * @Description：
 * @Author：jcshen
 * @Date：2023-08-12
 *
 * AggregateFunction的4个方法执行原理
 *  在Flink中AggregateFunction的四个方法在执行过程中会被转化为对应的内部Function对象，用于Flink的运行时执行计算
 *
 *  1、在数据输入流进入Flink的过程中，Flink会为每一个key创建对应的累加器。键值对流会按照键所在的组进行分区，然后把
 *     每一个组的所有元素分配到一个一个task slot中，并为每个key创建一个累加器。累加器的类型是任务的一个状态的函数，
 *     Flink根据累加器函数的类型来决定使用哪种累加器。
 *  2、当每个数据元素输入到累加器中时，add()方法会被调用。add()方法对输入的元素进行变换，然后更新累加器中的结果，
 *     返回新的结果给Flink的相应算子。
 *  3、在结果计算完毕后，getResult()方法将被调用，并将结果返回给Flink。
 *     最后，如果有多个累加器需要合并的情况，Flink会调用merge()方法将结果进行合并。
 *     通过这样的执行机制，AggregateFunction对象可以更加灵活快捷地处理数据。
 *  4、累加器的选择
 *      Flink支持两种类型的累加器：heap-based 和 incremental。
 *     当Flink创建累加器时，会根据AggregateFunction的类型来确定使用哪种类型的累加器。
 *     heap-based：累加器需要在内存中存储完整的所有元素，对于数据量较小情况，可以提供最好的性能；对于大数据量情况，可能会导致内存不足。
 *     incremental：累加器可以在输入元素上进行增量操作，并在内存中保存仅仅是必要的元素。
 *     可以处理更大数据量，并且在内存使用上更加高效。在使用增量累加器时，需要重写accumulate()和retract()方法。
 *  5、优点是简单实现聚合，缺点是不能输出key
 *
 */
public class GoodsAggFunc implements AggregateFunction<Behavior, Long, Long> {

    //初始化累加器，并将其初始化为默认值
    @Override
    public Long createAccumulator() {
        return 0L;
    }

    //将输入的元素添加到累加器，返回更新后的累加器，是来一条数据执行一次
    @Override
    public Long add(Behavior value, Long accumulator) {
        // 定义数据的添加逻辑
        return accumulator += 1;
    }

    //从累加器中提取操作的结果，窗口结束时执行一次
    @Override
    public Long getResult(Long accumulator) {
        // 定义计算结果的逻辑
        return accumulator;
    }

    //将两个累加器合并为一个新的累加器
    @Override
    public Long merge(Long aLong, Long acc1) {
        // 合并分区数据
        return null;
    }
}
